package hit.edu.zjc.Operator;


import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;
import hit.edu.zjc.parsers.ByteTupleParserFactory;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;

public class SourceSetTableMergeOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;
    private String[] FilePaths;//含有HashSet的二进制的文件路径
    // HashTable<Integer,HashSet<Integer>> SetTable;
    public SourceSetTableMergeOperatorDescriptor(IOperatorDescriptorRegistry spec,String[] files) {
        super(spec, 0, 1);  
        recordDescriptors[0]  = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        FilePaths=files;
    }

    @Override
    public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
    	 final ITupleParser parser =new ByteTupleParserFactory().createTupleParser(ctx);
    	return new AbstractUnaryOutputSourceOperatorNodePushable() {
        	
        	   /*
        	    *  将第二个SetTable，合并到第一SetTable上。
        	    * */
        	  public void MergeTwoSetTable(Hashtable<Integer,HashSet<Integer>> SetTable,Hashtable<Integer,HashSet<Integer>> otherSetTable)
        	  {
        		 Enumeration<Integer> e=otherSetTable.keys();
        		 while(e.hasMoreElements())
        		 {
        			 Integer key=e.nextElement();
        			 //HashSet<Integer>set =otherSetTable.get(key);
        			 //SetTable.put(key, set);
        			 HashSet<Integer>otherset=otherSetTable.get(key);
        			 HashSet<Integer>set=SetTable.get(key);
        			 if(set==null)
        			 {
        				 SetTable.put(key, otherset);
        			 }
        			 else
        			 {
        				 set.addAll(otherset);
        				 SetTable.put(key, set);
        			 }
        		 }
        	  }
        	
            @Override
            public void initialize() throws HyracksDataException {

            	final int filelength=FilePaths.length;
            	Hashtable<Integer,HashSet<Integer>> SetTable=(Hashtable<Integer,HashSet<Integer>>)RWObject.ReadObject(FilePaths[0]);
            	//把其他HashTable合并到第一个上。
            	for(int i=1;i<filelength;i++)
            	{
            		Hashtable<Integer,HashSet<Integer>> OtherSetTable=(Hashtable<Integer,HashSet<Integer>>)RWObject.ReadObject(FilePaths[i]);
            		MergeTwoSetTable(SetTable,OtherSetTable);
            	}
            	//合并结束。
      	
             writer.open();
             try
             	{
	     	    	byte[] SetTableBytes =ObjectConvert.ObjectToByteArray(SetTable);  		   
	     		   ByteArrayInputStream bais = new ByteArrayInputStream(SetTableBytes);	
	     	    	parser.parse(bais, writer);
	     	      bais.close();
	            
             	}
             catch(Exception e)
             	{
            	 	writer.fail();
            	 	e.printStackTrace();
             	}
             finally
             	{
            	 	writer.close();
             	}
            }
        };
    }
}